package com.netflexitysolutions.amazonws.sdb.orm.internal.operations;

import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.netflexitysolutions.amazonws.sdb.Domain;
import com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession;
import com.netflexitysolutions.amazonws.sdb.orm.filter.TypeFilter;
import com.netflexitysolutions.amazonws.sdb.orm.internal.util.GenericRecordComparator;
import com.netflexitysolutions.amazonws.sdb.orm.metadata.ItemMetadata;

public class FindAll<T> implements Callable<Set<T>> {

	private static Log logger = LogFactory.getLog(FindAll.class.getName());
	
	private SimpleDBSession session;
	private Class<T> clazz;
	private String[] orderBy;
	
	/**
	 * @param session
	 * @param clazz
	 */
	public FindAll(SimpleDBSession session, Class<T> clazz) {
		this.session = session;
		this.clazz = clazz;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.concurrent.Callable#call()
	 */
	public Set<T> call() throws Exception {
		// Obtain domain.
        Domain domain = session.getDomain(clazz);
        assert(domain != null);
        
        // Obtain metadata for this record type.
        ItemMetadata itemMetadata = session.getMetadata(clazz);
        assert(itemMetadata != null);
        
        List<T> records = session.cacheGetAll(clazz);
        if(records == null || records.isEmpty()){
        	logger.debug("Looking up all records in database for domain=" + domain.getName());
        	
	        // Select * records from database.
	        Select<T> select = new Select<T>(session, new TypeFilter<T>(clazz, session.getFactory().getMapping()), orderBy);
	        records = select.call();
	        
	        // Store results in cache.
	        for (T record : records) {
	        	session.cachePut(record);
			}
        }
        else{
        	// Apply orderBy.
            if(orderBy != null && orderBy.length > 0){
            	Collections.sort(records, new GenericRecordComparator<T>(session.getMetadata(clazz), orderBy));
            }
        }
        
        // Retrieve all records in parallel.
        /*ConcurrentFinder<T> finder = new ConcurrentFinder<T>(session, clazz, ids);
        List<T> records = finder.find();
        if(logger.isDebugEnabled()){
        	logger.debug("Found (" + records.size() + ") for domain " + domain.getName());
        }*/
        
        return new LinkedHashSet<T>(records);
	}

	/**
	 * @return the session
	 */
	public SimpleDBSession getSession() {
		return session;
	}

	/**
	 * @param session the session to set
	 */
	public void setSession(SimpleDBSession session) {
		this.session = session;
	}

	/**
	 * @return the clazz
	 */
	public Class<T> getClazz() {
		return clazz;
	}

	/**
	 * @param clazz the clazz to set
	 */
	public void setClazz(Class<T> clazz) {
		this.clazz = clazz;
	}
	
	/**
	 * @param orderBy the orderBy to set
	 */
	public void setOrderBy(String ... orderBy) {
		this.orderBy = orderBy;
	}
}
